package org.medusa.network.tcp;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.concurrent.atomic.AtomicLong;

import org.medusa.buffer.RingBuffer;
import org.medusa.core.Channel;
import org.medusa.core.Core;
import org.medusa.core.EventCallback;
import org.medusa.core.EventPump;
import org.medusa.log.ILog;
import org.medusa.network.Connector;
import org.medusa.packet.blob.BasePacket;

//1. runtime.Gosched();
//2. Packet ByteArray
//3.evp由AsyncConnector管理还是自管理 却取决于tcpconn构造时的evp对象是否为空
//4.release()重入问题
public abstract class BaseTcpAsyncConnector<PacketType extends BasePacket> extends
		Connector<PacketType> {

	private EventPump evp = null;
	private boolean self_evp = false;
	private ILog log;
	private String name = "unknow";
	private Socket fd = null;
	private String host = "localhost";
	private int port = 0;
	private int conn_timeout_ms;

	protected RingBuffer ringbuf = null;
	private byte[] rcvbuf = null;// = make([]byte, self.recv_buf_size/5)
	private int recv_buf_size = 0;
	private int send_buf_size = 0;
	// private int send_chan_num; //java don't need it

	// private HashMap<Integer, ProcCallback> handler = null;

	// private self.chan_join = make(chan int, 10) //java don't need it
	private Channel<PacketType> chan_sent = null;

	// private Packet half_packet_recv = null;

	// Event_ConnectionLose func(err error) //java don't need it
	// Event_UnknowPacketReceived func(pack *Packet) //java don't need it

	// event code define
	private long EVCODE_CONNECT_LOSE;
	private long EVCODE_CONNECT_PACKET_RECEIVED;

	// other
	private BaseTcpAsyncConnector<PacketType> self;
	private InputStream input = null;
	private OutputStream output = null;
	private Thread thread_recv = null;
	private Thread thread_sent = null;
	private Thread thread_idle = null;

	private int local_idle_ms;
	private static final int IDLE_MS_MIN_VALUE = 10000;// 10秒
	private AtomicLong heartbeat_begin_time_remote = new AtomicLong(0);
	private AtomicLong heartbeat_begin_time_local = new AtomicLong(0);

	public BaseTcpAsyncConnector(String name, ILog log, EventPump evp,
			int rcv_buf, int snd_buf, int idle_second) {

		this.evp = evp;
		if (this.evp == null) {
			this.evp = new EventPump(name);
			this.self_evp = true;
		}
		this.log = log;
		this.name = name;

		this.recv_buf_size = rcv_buf;
		this.send_buf_size = snd_buf;

		this.ringbuf = new RingBuffer(this.recv_buf_size, this.recv_buf_size/2);
		this.rcvbuf = new byte[this.recv_buf_size / 5];
		this.chan_sent = new Channel<PacketType>();

		this.local_idle_ms = idle_second * 1000;
		if (this.local_idle_ms < IDLE_MS_MIN_VALUE)
			this.local_idle_ms = IDLE_MS_MIN_VALUE;
		// 其他
		this.self = this;

	}

	// 心跳起搏器激活-远程
	private void remote_heartbeat_active() {
		Core.Gosched();
		this.heartbeat_begin_time_remote.set(System.currentTimeMillis());
	}

	// 远程心跳检查(对远程的心跳容忍度：3倍时间）
	private boolean remote_heartbeat_check() {
		Core.Gosched();
		boolean is_die = ((System.currentTimeMillis() - self.heartbeat_begin_time_remote
				.get()) > (this.local_idle_ms + 5000));
		if (is_die) {
			this.Error("$$$ remote silent time > " + (this.local_idle_ms + 5000)
					+ "ms, already death! $$$");// remote silent time more than
			self.Close();
		}
		return !is_die;
	}

	// 空闲时间重置-本地
	private void local_idle_reset() {
		Core.Gosched();
		this.heartbeat_begin_time_local.set(System.currentTimeMillis());
	}

	// 本地空闲检查(对本地的心跳容忍度：标准时间）
	private void local_idle_check() {
		Core.Gosched();
		boolean max_idle = ((System.currentTimeMillis() - this.heartbeat_begin_time_local
				.get()) >= this.local_idle_ms);
		if (max_idle) {
			this.Debug("local max idle!");
			this.event_callback_Idle();// 达到最大空闲时间,必须进行空闲处理
		}
		Core.Gosched();
		// return (System.currentTimeMillis()
		// - this.heartbeat_begin_time_local.get() < this.idle_ms);
	}

	public EventPump GetEvp() {
		return this.evp;
	}

	// protected abstract void event_callback_OnIdle();

	protected abstract void event_callback_ConnectionLose(Error err);

	protected abstract void event_callback_Idle();

	protected abstract void event_callback_pre_sent(PacketType pack);

	public long AddEventListener(EventCallback<Object[]> proc) {
		return this.evp.AddEventListener(proc);
	}

	public long AddDelayEventListener(long delay_ms, EventCallback<Object[]> proc) {
		return this.evp.AddDelayEventListener(delay_ms, proc);
	}

	public void DispatchEvent(long ev_code, Object... args) {
		this.evp.DispatchEvent(ev_code, args);
	}

	public void DelayRun(long delay_ms, EventCallback<Object[]> proc,
			Object... args) {
		// int ev_code = this.evp.AddDelayEventListener(delay_ms, proc);
		// this.evp.DispatchEvent(ev_code, args);
		this.evp.DelayRun(delay_ms, proc, args);
	}

	public long AddTimer(long interval_ms, long number,
			EventCallback<Object[]> proc) {
		return this.evp.AddTimerListener(interval_ms, number, proc);
	}

	public long AddTimer(long interval_ms, EventCallback<Object[]> proc) {
		return this.evp.AddTimerListener(interval_ms, proc);
	}
	
	public void ModifyTimerListener(long ev_code, long interval_ms, long number) {
		this.evp.ModifyTimerListener(ev_code, interval_ms, number);
	}

	public void RemoveTimer(long ev_code) {
		this.evp.RemoveTimerListener(ev_code);
	}

	@Override
	public Error Connect(String ip, int port, int timeout_ms) {
		Error err = null;
		if (this.is_running()) {
			return new Error("already connect (" + this.host + ":" + this.port
					+ ")");
		}
		this.chan_sent.Clear(); // golang don't need it
		this.host = ip;
		this.port = port;
		this.conn_timeout_ms = timeout_ms;

		try {
			this.fd = new Socket(/* ip, port */);
			this.fd.connect(new InetSocketAddress(ip, port), timeout_ms);

			this.fd.setReceiveBufferSize(this.recv_buf_size);
			this.fd.setSendBufferSize(this.send_buf_size);
			this.input = this.fd.getInputStream();
			this.output = this.fd.getOutputStream();
			this.set_state_running();

			// this.Success("ConnectionMade: (%v -> %v) err:%v",
			// self.GetLocalAddr(), self.GetRemoteAddr(), err);

			// event thread
			this.EVCODE_CONNECT_LOSE = this
					.AddEventListener(new EventCallback<Object[]>() {

						@Override
						public void execute(Object[] args) {
							Error err = (Error) args[0];
							self.event_callback_ConnectionLose(err);
							self.event_callback_exception_close();
						}

						@Override
						public void removed() {
							// TODO Auto-generated method stub
							
						}

					});

			this.EVCODE_CONNECT_PACKET_RECEIVED = this
					.AddEventListener(new EventCallback<Object[]>() {

						@Override
						public void execute(Object[] args) {
							@SuppressWarnings("unchecked")
							PacketType pack = (PacketType) args[0];
							self.event_callback_packet_received(pack);
						}

						@Override
						public void removed() {
							// TODO Auto-generated method stub
							
						}

					});

		} catch (UnknownHostException e) {
			//err = new Error(e.getMessage());
			this.release();
			return new Error(e.getMessage());

		} catch (IOException e) {
			//err = new Error(e.getMessage());
			this.release();
			return new Error(e.getMessage());
		}

		if (err == null)
			this.start_io_thread();

		return err;
	}

	private void start_io_thread() {

		this.local_idle_reset();
		this.remote_heartbeat_active();

		// recv thread
		this.thread_recv = new Thread(new Runnable() {//new Thread() {

			@Override
			public void run() {
				self.Error("recv thread is work!");
				while (self.is_running()) {
					try {
						int received = self.input.read(self.rcvbuf);
						if (received <= 0) {
							self.dispatch_event_exception_close(new Error(
									"remote disconnect!"));
							break;
						}
						// self.ringbuf.Write(self.rcvbuf[0:received])
						self.ringbuf.WriteBlob(self.rcvbuf, received);// java have
																	// no slice
						PacketType pack = null;// self.parse_packet();
						while (true) {
							pack = self.parse_packet();
							if (pack == null)
								break;
							self.remote_heartbeat_active();
							self.dispatch_event_packet_received(pack);

						}
						Core.Gosched();
					} catch (IOException e) {
						self.dispatch_event_exception_close(new Error(e
								.getMessage()));
						// e.printStackTrace();
						break;
					}
				}// end of : while
				self.Error("recv thread is free!");
			}
		
		});

		// sent thread
		this.thread_sent = new Thread(new Runnable() {//new Thread(){

			@Override
			public void run() {
				self.Error("sent thread is work!");
				while (self.is_running() && self.fd != null) {
					PacketType pack = null;
					pack = self.chan_sent.Read();
					if (pack == null) {
						self.dispatch_event_exception_close(new Error(
								"disconnect!"));
						break;
					} else {
						self.event_callback_pre_sent(pack);
						Error err = pack.Write2IO(self.output);
						if (err != null) {
							self.dispatch_event_exception_close(err);
							break;
						} else
							self.local_idle_reset();
					}
					Core.Gosched();
				}// end of : while
				self.Error("sent thread is free!");
			}

		});

		// idle thread
		this.thread_idle = new Thread(new Runnable() {//new Thread() {

			@Override
			public void run() {
				self.Error("idle thread is work!");
				while (self.is_running() && self.remote_heartbeat_check()) {
					self.local_idle_check();
					Core.Sleep(1000);
					Core.Gosched();
				}
				self.Error("idle thread is free!");
			}
		});

		this.thread_sent.start();
		this.thread_recv.start();
		this.thread_idle.start();
	}

	@Override
	public void Reconnect(int sleep_ms, String tip) {
		Error err = this.Connect(this.host, this.port, this.conn_timeout_ms);
		while (err != null) {
			Core.Sleep(sleep_ms);
			this.Warning(tip);
			err = this.Connect(this.host, this.port, this.conn_timeout_ms);
		}
	}

	@Override
	public void Close() {
		this.dispatch_event_exception_close(null);
	}

	@Override
	public void SendPacket(final PacketType pack) {
		if (pack != null) {
			this.chan_sent.Write(pack);
		}
	}

	@Override
	public boolean IsConnected() {
		return this.is_running();
	}

	public void Join() {
		try {
			if (this.thread_recv != null) {
				this.thread_recv.join();
			}
			if (this.thread_sent != null) {
				this.thread_sent.join();
			}

		} catch (InterruptedException e) {

		}
	}

	public void Error(Object... args) {
		if (this.log != null)
			this.log.Error(self.name, args);
	}

	public void Warning(Object... args) {
		if (this.log != null)
			this.log.Warning(self.name, args);
	}

	public void Success(Object... args) {
		if (this.log != null)
			this.log.Success(self.name, args);
	}

	public void Debug(Object... args) {
		if (this.log != null)
			this.log.Debug(self.name, args);
	}

	public void Stack(Object... args) {
		if (this.log != null)
			this.log.Stack(self.name, args);
	}

	public void Temp(Object... args) {
		if (this.log != null)
			this.log.Temp(self.name, args);
	}

	private void dispatch_event_packet_received(PacketType pack) {
		this.DispatchEvent(this.EVCODE_CONNECT_PACKET_RECEIVED, pack);
	}

	protected abstract void event_callback_packet_received(PacketType pack);

	private void dispatch_event_exception_close(Error err) {
		if (this.is_running())
			this.DispatchEvent(this.EVCODE_CONNECT_LOSE, err);
	}

	private void event_callback_exception_close() {
		this.release();
	}

	private void release() {
		if (this.self_evp)
			this.evp.Stop();
		if (this.set_state_stopped()) {
			this.thread_sent.interrupt();
			this.thread_recv.interrupt();
			// this.chan_sent.Write(null);
			if (self.fd != null) {
				try {
					this.input.close();
					this.output.close();
					this.fd.close();

					this.input = null;
					this.output = null;

					this.fd = null;

				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}

	}

	protected abstract PacketType parse_packet();

}
